package org.deng.learn.r2dbc.postgresql;

import io.r2dbc.postgresql.PostgresqlConnectionConfiguration;
import io.r2dbc.postgresql.PostgresqlConnectionFactory;
import io.r2dbc.postgresql.api.PostgresqlConnection;
import io.r2dbc.postgresql.api.PostgresqlResult;
import io.r2dbc.postgresql.api.PostgresqlStatement;
import io.r2dbc.spi.Connection;
import io.r2dbc.spi.Result;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * @author 邓力宾
 * @date 2021/2/23 15:37
 * @desc
 */
public class Main {
    static Map<String, String> options = new HashMap<>();
    static {
        options.put("lock_timeout", "10s");
    }
    static PostgresqlConnectionFactory connectionFactory = new PostgresqlConnectionFactory(PostgresqlConnectionConfiguration.builder()
            .host("192.168.2.177")
            .port(5432)  // optional, defaults to 5432
            .username("postgres")
            .password("denglibin")
            .database("denglibin")  // optional
            .options(options) // optional
            .build());
    public static void main(String[] args) throws InterruptedException {
       // insert();
        query();
    }

    /**
     * 添加
     */
    public static void insert(){
        Mono<PostgresqlConnection> mono = connectionFactory.create();
        Flux<PostgresqlResult> flux = mono.flatMapMany(connection -> {
            System.out.println("connection");
            return connection
                    .createStatement("INSERT INTO test.user (id, name, age, birthday) VALUES ($1, $2, $3, $4)")
                    .bind("$1", 12)
                    .bind("$2", "Walter2")
                    .bind("$3", 32)
                    .bind("$4", LocalDateTime.now())
                    .execute();
        });
        /*
        Mono<List<PostgresqlResult>> listMono = flux.collectList();
        listMono.block().forEach(reslut->reslut.getRowsUpdated().block());*/

       /* PostgresqlResult postgresqlResult = flux.blockLast();
        postgresqlResult.getRowsUpdated().block();*/

        flux.subscribe(reslut->reslut.getRowsUpdated().subscribe(i->System.out.println(i)));
        LockSupport.park(); //防止程序退出
    }

    /**
     * 查询
     */
    public static void query(){

        Mono<PostgresqlConnection> mono = connectionFactory.create();
        Flux<String> flux = mono.flatMapMany(connection -> {
            return connection
                    .createStatement("select name from test.company")
                    .execute()
                    .flatMap(r -> r.map((row, rowMetadata) -> {
                        StringBuilder sb = new StringBuilder();
                        rowMetadata.getColumnNames().forEach(column->{
                            sb.append(row.get(column, Object.class));
                            sb.append(" ");
                        });
                        return sb.toString();

                    }));
        });


        flux.subscribe(reslut-> System.out.println(reslut));
        LockSupport.park(); //防止程序退出
    }
}
